package basewsclient

import (
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"

	"gitee.com/851537/gbase"
	"github.com/cloudwego/hertz/pkg/common/hlog"
	"github.com/gorilla/websocket"
)

type Base struct {
	Name string

	Url   string
	Conn  *websocket.Conn
	debug bool

	//断线重连通道
	reconnectCh chan bool

	// 停止通道
	stopCh chan bool

	//心跳相关--------------------------------
	pingInterval int64                      // 心跳间隔时间，单位秒，0表示不开启心跳
	pingStopCh   chan bool                  // 心跳停止通道
	pingContent  string                     //ping内容
	pingType     int                        //ping类型，0:ping协议(默认)，1:发送文本
	pingHandler  func(appData string) error //收到ping的处理方法
	pongHandler  func(appData string) error //收到pong的处理方法
	//心跳相关--------------------------------

	mu sync.RWMutex

	// 写入锁，防止并发写入WebSocket连接
	writeMu sync.Mutex

	// RSV错误统计
	rsvErrorCount int64 // 使用原子操作

	OnConnected  func(child IWsClient)
	OnDisconnect func(child IWsClient, err error)
	OnMessage    func(child IWsClient, msg []byte, msgType ...int)
}

func (b *Base) SetName(name string) {
	b.Name = name
}

func (b *Base) SetUrl(url string) {
	b.Url = url
}

func (b *Base) SetDebug(status bool) {
	b.debug = status
}

func (b *Base) SetPingInterval(interval int64) {
	b.pingInterval = interval
}

func (b *Base) SetPingHandler(f func(appData string) error) {
	b.pingHandler = f
}
func (b *Base) SetPongHandler(f func(appData string) error) {
	b.pongHandler = f
}

func (b *Base) SetOnMessage(f func(child IWsClient, msg []byte, msgType ...int)) {
	b.OnMessage = f
}
func (b *Base) SetOnConnected(f func(child IWsClient)) {
	b.OnConnected = f
}
func (b *Base) SetOnDisconnect(f func(child IWsClient, err error)) {
	b.OnDisconnect = f
}

func (b *Base) SetPingType(pingType int) {
	b.pingType = pingType
}

func (b *Base) SetPingContent(pingContent string) {
	b.pingContent = pingContent
}

// 原子操作获取连接，避免高频场景下的锁竞争
func (b *Base) getConnAtomic() *websocket.Conn {
	return (*websocket.Conn)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&b.Conn))))
}

// 原子操作设置连接
func (b *Base) setConnAtomic(conn *websocket.Conn) {
	atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&b.Conn)), unsafe.Pointer(conn))
}

func (b *Base) GetConn() *websocket.Conn {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return b.Conn
}

func (b *Base) GetName() string {
	return b.Name
}

func (b *Base) GetUrl() string {
	return b.Url
}

func (b *Base) GetMsgObserverHub() *gbase.ObserverHub {
	return nil
}

func (b *Base) RegMsgObserver(observer gbase.IObserver) {
	hlog.Debugf("[%s]未实现RegMsgObserver方法", b.Name)
}

func (b *Base) Debug(format string, args ...any) {
	if b.debug {
		hlog.Debugf(format, args...)
	}
}

// GetRSVErrorCount 获取RSV错误统计次数
func (b *Base) GetRSVErrorCount() int64 {
	return atomic.LoadInt64(&b.rsvErrorCount)
}

// 检查WebSocket连接是否存在且处于活动状态，失败时触发重连
func (b *Base) CheckConn() bool {
	b.mu.RLock()
	conn := b.Conn
	b.mu.RUnlock()
	if conn == nil {
		b.triggerReconnect()
		return false
	}
	return true
}

// 触发重连
func (b *Base) triggerReconnect() {
	b.mu.Lock()
	reconnectCh := b.reconnectCh
	conn := b.Conn
	b.mu.Unlock()

	// 先关闭连接，强制让 readMessages 退出
	if conn != nil {
		conn.Close()
		// conn.LocalAddr().Network()
		b.Debug("[%s]强制关闭连接以触发重连", b.Name)
	}

	// 再发送重连信号
	if reconnectCh != nil {
		select {
		case reconnectCh <- true:
			b.Debug("[%s]连接检查失败，触发重连", b.Name)
		default:
			// 通道已满，不阻塞
			b.Debug("[%s]重连信号通道已满，跳过", b.Name)
		}
	}
}

// 检查是否收到停止信号
func (b *Base) checkRecvStopCh() bool {
	select {
	case <-b.stopCh:
		b.Debug("[%s]收到停止信号，退出重连循环", b.Name)
		return true
	default:
		// 继续执行连接逻辑
		return false
	}
}

func (b *Base) Connect(child IWsClient) {
	b.mu.Lock()
	// 初始化重连通道(确保只创建一次)
	if b.reconnectCh == nil {
		b.reconnectCh = make(chan bool, 1) // 改为缓冲通道
	}
	// 初始化心跳停止通道(确保只创建一次)
	if b.pingStopCh == nil {
		b.pingStopCh = make(chan bool, 1)
	}
	// 初始化停止重连通道(确保只创建一次)
	if b.stopCh == nil {
		b.stopCh = make(chan bool, 1)
	}
	b.mu.Unlock()

	for {
		// 检查是否收到停止信号，避免在connect和close同时并发调用，close方法失效
		if b.checkRecvStopCh() {
			return
		}
		// 创建独立的dialer实例，避免多个连接间的竞态条件
		dialer := &websocket.Dialer{
			HandshakeTimeout:  20 * time.Second,
			EnableCompression: false,                         // 禁用压缩扩展，解决 RSV 位错误和协议兼容性问题
			ReadBufferSize:    1024 * 128,                    // 128KB，平衡性能和内存使用
			WriteBufferSize:   1024 * 32,                     // 32KB
			Proxy:             websocket.DefaultDialer.Proxy, // 保持代理设置
		}

		// 设置更大的读写缓冲区，解决 bybit 等交易所大消息导致的 slice bounds out of range 错误
		// 根据实际需求选择合适的缓冲区大小：
		// - 32KB: 适合一般行情数据 (推荐)
		// - 128KB: 适合订阅多个交易对的大量数据
		// - 1MB: 适合极端情况下的超大消息
		// - 10MB: 内存消耗巨大，不推荐 (每连接占用20MB内存)

		b.Debug("[%s]start connect to[%s],timeout:%v,compression:%v,readBuf:%d,writeBuf:%d",
			b.Name, b.Url, dialer.HandshakeTimeout, dialer.EnableCompression,
			dialer.ReadBufferSize, dialer.WriteBufferSize)
		conn, _, err := dialer.Dial(b.Url, nil)
		if err != nil {
			b.Debug("[%s]Failed to connect to WebSocket server[%s], retrying in 5 seconds...,err:%s", b.Name, b.Url, err.Error())

			//连接失败事件
			if b.OnDisconnect != nil {
				b.OnDisconnect(child, err)
			}

			// 在等待重连期间也检查停止信号
			select {
			case <-b.stopCh:
				b.Debug("[%s]在重连等待期间收到停止信号，退出重连循环", b.Name)
				return
			case <-time.After(5 * time.Second):
				// 等待5秒后继续重连
			}
			continue
		}

		// 使用原子操作设置连接，同时保持锁保护其他字段
		b.mu.Lock()
		b.Conn = conn
		b.setConnAtomic(conn) // 原子操作设置，供高频读取使用
		b.mu.Unlock()

		//优化性能和稳定性-------------
		// 设置读取限制
		b.Conn.SetReadLimit(10 * 1024 * 1024) // 10MB，防止过大消息攻击

		// WebSocket 长连接不设置读取超时，避免正常情况下的 i/o timeout
		// 如果需要超时检测，应该通过心跳机制来实现
		// b.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))

		// 写入操作设置超时是安全的，防止写入阻塞
		// b.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))

		// 不启用压缩，避免协议兼容性问题
		// b.Conn.EnableWriteCompression(false) // 已在 dialer 中禁用

		//-------------------

		if b.pingHandler != nil {
			b.Conn.SetPingHandler(b.pingHandler)
		}

		if b.pongHandler != nil {
			b.Conn.SetPongHandler(b.pongHandler)
		}

		//连接成功事件
		if b.OnConnected != nil {
			b.OnConnected(child)
		}

		b.Debug("[%s]Connected to WebSocket server %s", b.Name, b.Url)

		// 使用WaitGroup确保goroutine安全
		var wg sync.WaitGroup

		// 启动读取消息的goroutine
		wg.Add(1)
		go func() {
			defer wg.Done()
			b.readMessages(child)
		}()

		// 如果设置了心跳间隔，启动心跳goroutine
		if b.pingInterval > 0 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				b.startPing()
			}()
		}

		// 等待重连信号（去重连）或停止信号（直接返回退出死循环）
		select {
		case <-b.reconnectCh:

		//continue
		case <-b.stopCh:
			b.Debug("[%s]收到停止信号，直接停止", b.Name)
			//其它清理已经在Close方法里完成
			return
		}
		//<-b.reconnectCh

		// 发送心跳停止信号
		if b.pingInterval > 0 {
			if b.pingStopCh != nil {
				select {
				case b.pingStopCh <- true:
					// 信号发送成功
				default:
					// 通道已满，不阻塞
				}
			}

		}

		wg.Wait() // 等待所有协程完全退出

		b.mu.Lock()
		if b.Conn != nil {
			if err = b.Conn.Close(); err == nil {
				b.Conn = nil
				b.setConnAtomic(nil) // 原子操作清理连接
			}
		}
		b.mu.Unlock()
		// if b.OnDisconnect != nil {
		// 	b.OnDisconnect(child, errors.New("连接失败，尝试重连"))
		// }
		hlog.Debugf("[%s]Disconnected from WebSocket server, attempting to reconnect...", b.Name)
	}
}

func (b *Base) readMessages(child IWsClient) {
	defer func() {
		if r := recover(); r != nil {
			hlog.Errorf("[%s]Panic in readMessages: %+v", b.Name, r)
		}
		// 确保在任何情况下都发送重连信号
		select {
		case b.reconnectCh <- true:
			// 信号发送成功
		default:
			// 通道已满，不阻塞
		}
	}()
	for {
		// 使用原子操作获取连接，避免高频场景下的锁竞争
		conn := b.getConnAtomic()

		if conn == nil {
			b.Debug("[%s]连接已为nil，退出读取循环", b.Name)
			return
		}

		msgType, message, err := conn.ReadMessage()
		if err != nil {
			hlog.Errorf("[%s]Error reading message:%s", b.Name, err.Error())

			// 检查是否是轻微的协议错误（仅在特定条件下才忽略）
			errStr := err.Error()
			if (strings.Contains(errStr, "RSV") ||
				strings.Contains(errStr, "bad opcode") ||
				strings.Contains(errStr, "continuation after FIN") ||
				strings.Contains(errStr, "bad MASK")) &&
				!strings.Contains(errStr, "repeated read") {

				// 原子操作增加RSV错误计数
				count := atomic.AddInt64(&b.rsvErrorCount, 1)
				b.Debug("[%s]检测到协议错误[第%d次]，但连接可能仍可用，尝试继续: %s", b.Name, count, errStr)

				// 如果RSV错误过于频繁，记录警告
				if count%100 == 0 {
					hlog.Warnf("[%s]RSV协议错误已发生%d次，建议检查服务器兼容性", b.Name, count)
				}

				continue
			}

			// 对于 "repeated read" 或其他严重错误，必须退出
			if strings.Contains(errStr, "repeated read") {
				hlog.Errorf("[%s]检测到重复读取错误，连接已失效，退出读取循环", b.Name)
			}

			if b.OnDisconnect != nil {
				b.OnDisconnect(child, fmt.Errorf("读消息失败，尝试重连 %s", err.Error()))
			}

			//网络断线等严重错误，触发重连
			return // 退出读取循环，触发重连
		}

		// 检查消息大小
		// if len(message) > 1024*1024 { // 1MB
		// 	hlog.Warnf("[%s]Received large message: %d bytes", b.Name, len(message))
		// }

		b.Debug("[%s]Received msgType:%d message: %s\n", b.Name, msgType, message)

		if b.OnMessage != nil {
			b.OnMessage(child, message, msgType)
		}
	}
}

// 发送消息到服务器
func (b *Base) Send(child IWsClient, msg []byte) error {
	b.mu.RLock()
	conn := b.Conn
	b.mu.RUnlock()

	b.Debug("[%s]wsclient.Send msg:%s", b.Name, string(msg))

	if conn == nil {
		hlog.Debugf("[%s]wsclient.Send failed,due to conn closed,msg:%s", b.Name, string(msg))
		return websocket.ErrCloseSent //ErrNotConnected
	}

	// 使用写入锁防止并发写入
	b.writeMu.Lock()
	defer b.writeMu.Unlock()

	return conn.WriteMessage(websocket.TextMessage, msg)
}

// 开启发送心跳包
func (b *Base) startPing() {
	b.Debug("[%s]startPing,interval:%d,content:%s,type:%d", b.Name, b.pingInterval, b.pingContent, b.pingType)
	ticker := time.NewTicker(time.Duration(b.pingInterval) * time.Second)
	defer ticker.Stop()
	if b.pingContent == "" {
		b.pingContent = "ping"
	}

	for {
		select {
		case <-ticker.C:
			b.mu.RLock()
			conn := b.Conn
			b.mu.RUnlock()

			if conn != nil {
				b.Debug("[%s]发送ping content:%s type:%d", b.Name, b.pingContent, b.pingType)
				// 使用写入锁防止并发写入
				b.writeMu.Lock()
				if b.pingType == PingTypeStd {
					err := conn.WriteControl(websocket.PingMessage, []byte(b.pingContent), time.Now().Add(10*time.Second))
					if err != nil {
						b.Debug("[%s]发送ping失败 %s", b.Name, err.Error())

					} else {
						b.Debug("[%s]发送ping成功", b.Name)
					}
				} else {
					err := conn.WriteMessage(websocket.TextMessage, []byte(b.pingContent))
					if err != nil {
						b.Debug("[%s]发送ping失败 %s", b.Name, err.Error())
					} else {
						b.Debug("[%s]发送ping成功", b.Name)
					}
				}
				b.writeMu.Unlock()
			}
		case <-b.pingStopCh: //收到停止信号,结束本协程
			b.Debug("[%s]Ping stopped", b.Name)
			return
		}
	}
}

// 关闭websocket连接,关闭后不再重连
func (b *Base) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()

	// 发送停止重连信号(即使Conn为nil也要发送，因为Connect循环可能还在运行)
	if b.stopCh != nil {
		select {
		case b.stopCh <- true:
			// 信号发送成功
		default:
			// 通道已满，不阻塞
		}
		close(b.stopCh)
		b.stopCh = nil
	}

	if b.Conn != nil {
		// 关闭心跳
		if b.pingStopCh != nil {
			select {
			case b.pingStopCh <- true:
				// 信号发送成功
			default:
				// 通道已满，不阻塞
			}
			close(b.pingStopCh)
			b.pingStopCh = nil
		}

		// 关闭重连通道
		if b.reconnectCh != nil {
			select {
			case b.reconnectCh <- true:
				// 发送关闭信号
			default:
				// 通道已满，不阻塞
			}
			close(b.reconnectCh)
			b.reconnectCh = nil
		}

		// 关闭websocket连接
		// 使用写入锁防止并发写入
		b.writeMu.Lock()
		err := b.Conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
		b.writeMu.Unlock()
		if err != nil {
			b.Debug("[%s]写入关闭消息失败: %v", b.Name, err)
		}

		err = b.Conn.Close()
		if err != nil {
			b.Debug("[%s]关闭连接失败: %v", b.Name, err)
		}

		b.Conn = nil
		b.setConnAtomic(nil) // 原子操作清理连接

		b.Debug("[%s]WebSocket连接已关闭", b.Name)
	}
}
